package com.example.utils;

import com.jcraft.jsch.IO;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.math.BigInteger;
import java.text.MessageFormat;
import java.util.*;

@Slf4j
public class HbaseService {

    private Configuration conf = null;
    private Connection connection = null;

    public HbaseService(Configuration conf1) {
        this.conf = conf1;
        try {
            connection = ConnectionFactory.createConnection(conf);
        } catch (IOException e) {
            log.error("获取HBase连接失败");
        }
    }
    private void close(Admin admin, ResultScanner rs, Table table){
        if(admin != null){
            try {
                admin.close();
            } catch (IOException e) {
                log.error("关闭Admin失败",e);
            }
        }

        if(rs != null){
            rs.close();
        }

        if(table != null){
            try {
                table.close();
            } catch (IOException e) {
                log.error("关闭Table失败",e);
            }
        }
    }
    /**
     * 创建命名空间
     * */
    public void createNamespace(String namespace) {
        Admin admin = null;
        try{
            //获取admin,admin连接是轻量级，不是线程安全的，不推荐池化或者缓存这个连接
            admin = connection.getAdmin();
            //调用方法创建命名空间
            //创建命名空间描述建造者 =>设计师
            NamespaceDescriptor.Builder builder = NamespaceDescriptor.create(namespace);
            //给命名空间添加需求
            builder.addConfiguration("user","xiaoming");
            //使用builder构造出对应的添加完参数的对象，完成创建
            admin.createNamespace(builder.build());

            admin.close();

        }catch (IOException e){
            e.printStackTrace();
        }


    }
    /**
     * 判断表是否存在
     * */
    public boolean tableExists(String namespace,String tableName) {
        Admin admin = null;
        boolean exists = false;
        try{
            admin = connection.getAdmin();
            //使用方法判断表格是否存在
            exists = admin.tableExists(TableName.valueOf(namespace, tableName));
            admin.close();

        } catch (IOException e) {
            e.printStackTrace();
        }
        return exists;
    }

    /**
     * 创建表
     * @author
     * @date
     * @since 1.0.0
     * @param tableName 表名
     * @param columnFamilies 列族名
     * @return void
     */
    public boolean createTable(String namespace,String tableName, List<String> columnFamilies) {
        boolean exists = tableExists(namespace, tableName);
        if(exists){
            log.error("表格已经存在!");
           return false;
        }
        Admin admin = null;
        try {
            admin = connection.getAdmin();

            ArrayList<ColumnFamilyDescriptor>  familyDescriptors = new ArrayList<>(columnFamilies.size());
            if(familyDescriptors.size() == 0){
                log.error("创建表格至少有一个列族");
                return false;
            }

            columnFamilies.forEach(columnFamily -> {
                familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(columnFamily)).build());
            });

            TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(namespace,tableName))
                    .setColumnFamilies(familyDescriptors)
                    .build();

            if(admin.tableExists(TableName.valueOf(tableName))) {
                log.debug("table exists");
            }else{
                admin.createTable(tableDescriptor);
                log.debug("create table success!");
            }
        }catch (IOException e){
            log.error("create table fail!");
            return false;
        }finally {
            close(admin,null,null);
        }
        return true;
    }

    /**
     * 预分区创建表
     * @param tableName 表名
     * @param columnFamily 列族名的集合
     * @param splitKeys 预分期region
     * @return 是否创建成功
     */
    public boolean createTableBySplitKeys(String tableName, List<String> columnFamily, byte[][] splitKeys) {
        Admin admin = null;
        try {
            if(StringUtils.isBlank(tableName) || columnFamily == null
             || columnFamily.size() == 0){
                log.error("===parameters tableName|columnFamily should not be null or empty");
                return false;
            }
            admin = connection.getAdmin();
            if(admin.tableExists(TableName.valueOf(tableName))){
                return true;
            }else {
                List<ColumnFamilyDescriptor> familyDescriptors = new ArrayList<>(columnFamily.size());
                columnFamily.forEach(cf ->{
                    familyDescriptors.add(ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(cf)).build());
                });
                TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
                        .setColumnFamilies(familyDescriptors)
                        .build();

                //指定splitkeys
                admin.createTable(tableDescriptor,splitKeys);
                log.info("====create table" + tableName
                        +"Success!columnFamilies:" + columnFamily.toString()+"===");
            }
        } catch (IOException e) {
            log.info("create table fail!"+e.getMessage());
            return false;
        }finally {
            close(admin,null,null);
        }
        return true;
    }

    /**
     * 自定义获取分区splitKeys
     * */
    public byte[][] getSplitKeys(String[] keys){
        if (keys == null) {
            keys = new String[]{"1|", "2|", "3|", "4|", "5|", "6|", "7|", "8|", "9|"};
        }
        byte[][] splitKeys = new byte[keys.length][];
        //升序排序
        TreeSet<byte[]> rows = new TreeSet<byte[]>(Bytes.BYTES_COMPARATOR);
        for (String key : keys) {
            rows.add(Bytes.toBytes(key));
        }
        Iterator<byte[]> rowkeyIter = rows.iterator();
        int i = 0;
        while(rowkeyIter.hasNext()){
            byte[] next = rowkeyIter.next();
            rowkeyIter.remove();
            splitKeys[i] = next;
            i++;
        }
        return splitKeys;
    }
    /**
     * 按startkey和endkey,分区数获取分区
     * */
    public static byte[][] getHexSplites(String startKey, String endKey,int numRegions){
        byte[][] splits = new byte[numRegions - 1][];
        BigInteger lowestKey = new BigInteger(startKey, 16);
        BigInteger highestKey = new BigInteger(endKey, 16);
        BigInteger range = highestKey.subtract(lowestKey);
        BigInteger regionIncrement = range.divide(BigInteger.valueOf(numRegions));
        lowestKey = lowestKey.add(regionIncrement);
        for (int i = 0; i < numRegions - 1; i++) {
            BigInteger key = lowestKey.add(regionIncrement.multiply(BigInteger.valueOf(i)));
            byte[] b = String.format("%016x", key).getBytes();
            splits[i] = b;
        }
        return splits;
    }

    /**
     * 获取table
     * */
    private Table getTable(String tableName) throws IOException {
        return connection.getTable(TableName.valueOf(tableName));
    }

    /**
     * 查询库中的所有表名
     * */
    public List<String> getAllTableNames() {
        List<String> result = new ArrayList<>();
        Admin admin = null;
        try{
            admin = connection.getAdmin();
            TableName[] tableNames = admin.listTableNames();
            for (TableName tableName : tableNames) {
                result.add(tableName.getNameAsString());
            }
        }catch (IOException e){
            log.error("getAllTableNames fail!",e);
        }finally {
            close(admin,null,null);
        }
        return result;
    }

    /**
     * 通过表名以及过滤条件查询数据
     * */
    private Map<String,Map<String,String>> queryData(String tableName,Scan scan) throws IOException {
        //<rowkey,对应的行数据>
        HashMap<String, Map<String,String>> result = new HashMap<>();

        ResultScanner rs = null;
        //获取表
        Table table = null;
        try{
            table = getTable(tableName);
            rs = table.getScanner(scan);
            for (Result r : rs) {
                HashMap<String, String> columnMap = new HashMap<>();
                String rowKey = null;
                for (Cell cell : r.listCells()) {
                    if (rowKey == null) {
                        rowKey = Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
                    }
                    columnMap.put(Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()), Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
                }
                if (rowKey != null) {
                    result.put(rowKey, columnMap);
                }

            }
        }catch (IOException e){
            log.error(MessageFormat.format("遍历查询指定表中的所有数据失败，tableName:{0}",tableName),e);
        }finally {
            close(null,rs,table);
        }
        return result;
    }


    /**
     * 遍历查询指定表中所有数据
     * */
    public Map<String, Map<String,String>> getResultScanner(String tableName) throws IOException {
        Scan scan = new Scan();
        return this.queryData(tableName,scan);
    }
    /**
     * 通过行前缀过滤器查询数据
     * */
    public Map<String,Map<String,String>> getResultScannerPrefixFilter(String tableName, String prefix) throws IOException {
        Scan scan = new Scan();

        if(StringUtils.isNotBlank(prefix)){
            Filter filter = new PrefixFilter(Bytes.toBytes(prefix));
            scan.setFilter(filter);
        }
        return this.queryData(tableName,scan);
    }

    /**
     * 通过列前缀过滤器查询数据
     * */
    public Map<String,Map<String,String>> getResultScannerColumnPrefixFilter(String tableName, String prefix) throws IOException {
        Scan scan = new Scan();
        if(StringUtils.isNotBlank(prefix)){
            Filter filter = new ColumnPrefixFilter(Bytes.toBytes(prefix));
            scan.setFilter(filter);
        }
        return this.queryData(tableName,scan);
    }

    /**
     * 查询行键中包含特殊字符的数据
     * */
    public Map<String,Map<String,String>> getResultScannerRowFilter(String tableName,String keyword) throws IOException {
        Scan scan = new Scan();

        if(StringUtils.isNotBlank(keyword)){
            Filter filter = new RowFilter(CompareOperator.GREATER_OR_EQUAL, new SubstringComparator(keyword));
            scan.setFilter(filter);
        }
        return this.queryData(tableName,scan);
    }

    /**
     * 查询列名中包含特定字符的数据
     * */
    public Map<String,Map<String,String>> getResultScannerQualifierFilter(String tableName, String keyword) throws IOException {
        Scan scan = new Scan();

        if(StringUtils.isNotBlank(keyword)){
            Filter filter = new QualifierFilter(CompareOperator.GREATER_OR_EQUAL, new SubstringComparator(keyword));
            scan.setFilter(filter);
        }
        return this.queryData(tableName,scan);
    }

    /**
     * 根据tableName和rowKey精确查询一行的数据
     * */
    private Map<String,String> getRowData(String tableName,String rowkey) throws IOException {
        Map<String,String> result = new HashMap<>();

        Get get = new Get(Bytes.toBytes(rowkey));
        //获取表
        Table table = null;
        try{
            table = getTable(tableName);
            Result hTableResult = table.get(get);
            if(hTableResult != null && !hTableResult.isEmpty()){
                for (Cell cell : hTableResult.listCells()) {
                    result.put(Bytes.toString(cell.getQualifierArray(),cell.getQualifierOffset(),cell.getQualifierLength()),
                            Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getQualifierLength()));

                }
            }
        }catch (IOException e){
            log.error(MessageFormat.format("查询一行的数据失败,tableName:{0},rowkey:{1}",tableName,rowkey),e);
        }finally {
            close(null,null,table);
        }
        return result;
    }

    /**
     * 根据tableName,rowKey,familyName,column查询指定单元格的数据
     * */
    public String getColumnValue(String tableName,String rowKey,String familyName,String columnName){
        String str = null;
        Get get = new Get(Bytes.toBytes(rowKey));
        //获取表
        Table table = null;
        try{
            table = getTable(tableName);
            Result result = table.get(get);
            if(result != null && !result.isEmpty()){
                Cell cell = result.getColumnLatestCell(Bytes.toBytes(familyName), Bytes.toBytes(columnName));
                if(cell != null){
                    str = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
                }
            }
        } catch (IOException e) {
            log.error(MessageFormat.format("查询指定单元格的数据失败,tableName:{0},rowKey:{1},familyName:{2},columnName:{3}",
                    tableName,rowKey,familyName,columnName),e);
        }finally {
            close(null,null,table);
        }
        return str;
    }
    /**
     * 根据tableName,rowkey,familyName,column查询指定单元格多个版本的数据
     * */
    public List<String> getColumnValuesByVersion(String tableName,String rowKey,String familyName,String columnName,int versions){
        //返回数据
        List<String> result = new ArrayList<>(versions);

        //获取表
        Table table = null;
        try{
             table = getTable(tableName);
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
            //读取多少个版本
            get.readVersions(versions);
            Result hTableResult = table.get(get);
            if(hTableResult != null && !hTableResult.isEmpty()){
                for (Cell cell : hTableResult.listCells()) {
                    result.add(Bytes.toString(cell.getValueArray(),cell.getValueOffset(),cell.getValueLength()));
                }
            }
        } catch (IOException e) {
            log.error(MessageFormat.format("查询指定单元格多个版本的数据失败，,tableName:{0},rowKey:{1},familyName:{2},columnName:{3}",
                    tableName,rowKey,familyName,columnName),e);
        }finally {
            close(null,null,table);
        }
        return result;
    }

    /**
     * 为表添加 or 更新数据
     * */
    public void putData(String tableName,String rowKey,String familyName,String[] columns,String[] values){
        //获取表
        Table table = null;
        try{
             table = getTable(tableName);

             putData(table,rowKey,tableName,familyName,columns,values);
        }catch (Exception e){
            log.error(MessageFormat.format("为表添加 or 更新数据失败,tableName:{0},rowKey:{1},familyName:{2}"
             ,tableName,rowKey,familyName),e);
        }finally {
            close(null,null,table);
        }
    }

    public void putData(Table table,String rowKey,String tableName,String familyName,String[] columns,String[] values){
        try{
            Put put = new Put(Bytes.toBytes(rowKey));
            if(columns != null && values != null && columns.length == values.length){
                for(int i=0;i<columns.length;i++){
                    if(columns[i] != null && values[i] != null){
                        put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columns[i]),Bytes.toBytes(values[i]));
                    }else{
                        throw new NullPointerException(MessageFormat.format("列名和列数据不能为空，column:{0},value:{1}"
                        ,columns[i],values[i]));
                    }
                }
            }
            table.put(put);
            log.debug("putData add or update data Success,rowkey:"+ rowKey);
            table.close();
        }catch (Exception e){
            log.error(MessageFormat.format("为表添加 or 更新数据失败，tableName:{0},rowKey:{1},familyName:{2}"
            ,tableName,rowKey,familyName),e);
        }
    }
    /**
     * 为表的某个单元格赋值
     * */
    public void setColumnValue(String tableName,String rowKey,String familyName,String column1,String value1){
        Table table = null;
        try{
            //获取表
             table = getTable(tableName);
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(column1),Bytes.toBytes(value1));
            table.put(put);
            log.debug("add data Success!");
        }catch (IOException e){
            log.error(MessageFormat.format("为表的某个单元格赋值失败，tableName:{0},rowKey:{1},familyName:{2},column:{3}"
            ,tableName,rowKey,familyName,column1),e);
        }finally {
            close(null,null,table);
        }
    }
    /**
     * 删除指定的单元格
     * */
    public boolean deleteColumn(String tableName,String rowKey,String familyName,String columnName){
        Table table = null;
        Admin admin = null;
        try{
             admin = connection.getAdmin();
             if(admin.tableExists(TableName.valueOf(tableName))){
                  table = getTable(tableName);
                 Delete delete = new Delete(Bytes.toBytes(rowKey));
                 //设置待删除的列
                 delete.addColumn(Bytes.toBytes(familyName),Bytes.toBytes(columnName));
                 table.delete(delete);
                 log.debug(MessageFormat.format("familyName:{0},columnName:{1} is deleted",familyName,columnName));
             }
        }catch (IOException e){
            log.error(MessageFormat.format("删除指定的列失败，tableName:{0},rowKey:{1},familyName:{2},column:{3}"
            ,tableName,rowKey,familyName,columnName),e);
            return false;
        }finally {
            close(admin,null,table);
        }
        return true;
    }

    /**
     * 根据rowKey删除指定的行
     * */
    public boolean deleteRow(String tableName,String rowKey){
        Table table = null;
        Admin admin = null;
        try{
            admin = connection.getAdmin();

            if(admin.tableExists(TableName.valueOf(tableName))){
                table = getTable(tableName);
                Delete delete = new Delete(Bytes.toBytes(rowKey));

                table.delete(delete);
                log.debug(MessageFormat.format("row({0}) is deleted!",rowKey));
            }
        }catch (IOException e){
            log.error(MessageFormat.format("删除指定的行失败，tableName:{0},rowKey:{1}"
            ,tableName,rowKey),e);
            return false;
        }finally {
            close(admin,null,table);
        }
        return true;
    }

    /**
     * 根据columnFamily删除指定的列族
     * */
    public boolean deleteColumnFamily(String tableName,String columnFamily){
        Admin admin = null;
        try{
            admin = connection.getAdmin();

            if(admin.tableExists(TableName.valueOf(tableName))){
                admin.deleteColumnFamily(TableName.valueOf(tableName),Bytes.toBytes(columnFamily));
                log.debug(MessageFormat.format("columnFamily({0}) is deleted!",columnFamily));
            }
        }catch (IOException e){
            log.error(MessageFormat.format("删除指定的列族失败，tableName:{0},columnFamily:{1}"
            ,tableName,columnFamily),e);
            return false;
        }finally {
            close(admin,null,null);
        }
        return true;
    }

    /***
     * 删除表
     */
    public boolean deleteTable(String tableName){
        Admin admin = null;
        try {
            admin = connection.getAdmin();
            if(admin.tableExists(TableName.valueOf(tableName))){
                admin.disableTable(TableName.valueOf(tableName));
                admin.deleteTable(TableName.valueOf(tableName));
                log.debug(tableName + " is deleted!");
            }
        }catch (IOException e){
            log.error(MessageFormat.format("删除指定的表失败，tableName:{0}"
            ,tableName),e);
            return false;
        }finally {
            close(admin,null,null);
        }
        return true;
    }





}
